package com.wangzaiplus.test.juc.thread;


import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * @ClassName TaskDisposeUtils
 * @Description: TODO
 * @Author wgc
 * @Date 2020/06/29 11:43
 * @Version V1.0
 **/
public class TaskDisposeUtils {
    //并行线程数
    public static final int POOL_SIZE;

    static {
        POOL_SIZE = Integer.max(Runtime.getRuntime().availableProcessors(), 5);
    }

    //并行处理和并行结束
    public static <T> void dispose(List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        dispose(true, POOL_SIZE, taskList, consumer);
    }

    private static <T> void dispose(boolean moreThread, int poolSize, List<T> taskList, Consumer<T> consumer) throws InterruptedException {
        if (CollectionUtils.isEmpty(taskList)) {
            return;
        }
        if(moreThread&&poolSize >1){
            poolSize=Math.min(poolSize,taskList.size());
            ExecutorService executorService=null;
           try {
               executorService = Executors.newFixedThreadPool(poolSize);
               CountDownLatch countDownLatch=new CountDownLatch(taskList.size());
               for (T t:taskList){
                   executorService.execute(()->{
                       try {
                           consumer.accept(t);
                       }finally {
                           countDownLatch.countDown();
                       }
                   });
               }
               countDownLatch.await();
           }finally {
                if(executorService!=null){
                    executorService.shutdownNow();
                }
           }
        }else{
            for (T t:taskList){
                consumer.accept(t);
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        List<Integer> list = Stream.iterate(1, a -> a + 1).limit(50).collect(Collectors.toList());
        TaskDisposeUtils.dispose(list,item->{
            long startTime=System.currentTimeMillis();
            try {
                TimeUnit.SECONDS.sleep(item);
                long endTime=System.currentTimeMillis();
                System.out.println(System.currentTimeMillis()+",任务"+item+" 执行完毕,耗时"+(endTime-startTime));
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        System.out.println(list+"全部执行完毕");
    }
}
